package com.example.demo.work;

import com.example.demo.handler.SendHandler;
import com.example.demo.maintainer.GatewayOfficer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 单个通道操作单元
 * Created by guoyangqiao on 10/05/2017.
 */
public class GatewayOperator {
    private static final Logger LOGGER = LoggerFactory.getLogger(GatewayOperator.class);
    //这个通道的ID
    private int gatewayId;
    //获取通道执行权的时候使用的占用ID
    private String privilegeSerId;
    //这个通道最大的线程数,默认单线程
    private int maxConcurrentThreadNum;
    //通道是否可以被执行发送
    private boolean sendHandleAllowed;
    //通道是否在发送时间
    private boolean sendTimeAllowed;
    //最后一次维持通道在本地的时刻,用于保证当前通道的执行权在本应用
    private long lastUpdateTime;

    //调用通道实现的相关配置
    private String sendConfig;
    //每次获取子任务的最大数量
    private int maxTaskFetchSize;

    //每收到一次消息+1,通道每发送一批任务获取一次,如果通道发送完counter和这个值一样则没有更多任务,发送线程退出
    private final AtomicLong acceptedWorkCounter = new AtomicLong(0);
    //当前正在用这个通道的工作线程数
    private final AtomicLong currentRunningTreads = new AtomicLong(0);



    public void addAcceptedWorkCounter() {
        this.acceptedWorkCounter.incrementAndGet();
    }

    public void setMaxConcurrentThreadNum(int maxConcurrentThreadNum) {
        this.maxConcurrentThreadNum = maxConcurrentThreadNum;
    }

    public void setSendHandleAllowed(boolean sendHandleAllowed) {
        this.sendHandleAllowed = sendHandleAllowed;
    }

    public void setSendTimeAllowed(boolean sendTimeAllowed) {
        this.sendTimeAllowed = sendTimeAllowed;
    }

    public void setLastUpdateTime() {
        this.lastUpdateTime = System.currentTimeMillis();
    }

    public void setSendConfig(String sendConfig) {
        this.sendConfig = sendConfig;
    }

    public void setMaxTaskFetchSize(int maxTaskFetchSize) {
        this.maxTaskFetchSize = maxTaskFetchSize;
    }

    public int getGatewayId() {
        return gatewayId;
    }

    public String getPrivilegeSerId() {
        return privilegeSerId;
    }

    public int getMaxConcurrentThreadNum() {
        return maxConcurrentThreadNum;
    }

    public boolean isSendHandleAllowed() {
        return sendHandleAllowed;
    }

    public boolean isSendTimeAllowed() {
        return sendTimeAllowed;
    }

    public long getLastUpdateTime() {
        return lastUpdateTime;
    }

    public String getSendConfig() {
        return sendConfig;
    }

    public int getMaxTaskFetchSize() {
        return maxTaskFetchSize;
    }

    public long getAcceptedWorkCounter() {
        return acceptedWorkCounter.get();
    }

    public AtomicLong syncBeforeSubmitThread() {
        return currentRunningTreads;
    }

    public long getCurrentRunningTreadNum() {
        return currentRunningTreads.get();
    }

    public void increaseCurrentRunningTreadNum() {
        currentRunningTreads.incrementAndGet();
    }

    public long decreaseCurrentRunningTreads() {
        return currentRunningTreads.decrementAndGet();
    }

    /**
     * @param gatewayId                      通道id
     * @param privilegeSerId                 占用值
     * @param privilegeLastUpdateMillSeconds 更新时间
     * @param sendHandleAllowed              通道有效
     * @param sendTimeAllowed                发送时间有效
     * @param sendConfig                     发送配置
     * @param maxTaskFetchSize               最大获取量
     * @param maxConcurrentThreadNum         最大并行线程数
     * @return
     */
    public static GatewayOperator newInstance(int gatewayId,
                                              String privilegeSerId,
                                              long privilegeLastUpdateMillSeconds,
                                              boolean sendHandleAllowed,
                                              boolean sendTimeAllowed,
                                              String sendConfig,
                                              int maxTaskFetchSize,
                                              int maxConcurrentThreadNum) {
        GatewayOperator instance = new GatewayOperator();
        instance.gatewayId = gatewayId;
        instance.privilegeSerId = privilegeSerId;
        instance.lastUpdateTime = privilegeLastUpdateMillSeconds;
        instance.sendHandleAllowed = sendHandleAllowed;
        instance.sendTimeAllowed = sendTimeAllowed;
        instance.sendConfig = sendConfig;
        instance.maxTaskFetchSize = maxTaskFetchSize;
        instance.maxConcurrentThreadNum = maxConcurrentThreadNum;
        return instance;
    }

    /** 内部类 调用handle处理业务*/
    public static class GatewayWorkerRunner implements Runnable {
        private GatewayOperator operator;

        public GatewayWorkerRunner(GatewayOperator operator) {
            this.operator = operator;
        }

        @Override
        public void run() {
            long acceptedWorkCounter = operator.getAcceptedWorkCounter();
            int gatewayId = operator.getGatewayId();
            LOGGER.info("开始执行通道{}的发送任务", gatewayId);
            try {
                SendHandler sendHandler = (SendHandler) Class.forName("com.example.demo.handler.imp.BaseHandler").newInstance();
                sendHandler.handle(operator);
            } catch (Exception e) {
                LOGGER.error("通道{}实际发送异常", gatewayId, e);
            }
            //调用这个函数意味着这个工作线程已经结束发送
            GatewayOfficer.afterThreadWorkFinished(operator, acceptedWorkCounter);
        }
    }

    @Override
    public String toString() {
        return "GatewayOperator{" +
                "gatewayId=" + gatewayId +
                ", maxConcurrentThreadNum=" + maxConcurrentThreadNum +
                ", sendHandleAllowed=" + sendHandleAllowed +
                ", sendTimeAllowed=" + sendTimeAllowed +
                ", sendConfig='" + sendConfig + '\'' +
                ", maxTaskFetchSize=" + maxTaskFetchSize +
                '}';
    }
}
